package LogThread;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

import org.apache.commons.lang3.StringUtils;

import Dao.FilterVisiteDao;

import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.Session;

public class Connector {
	static String hostname = null;
	static String username = null;
	static String password = null;

//	public static void main(String[] arg0) {
//		try {
//			toFileAppend();
//			toMapReduce();
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
//	}

	public static void toFileAppend() throws Exception {

		File file = null;
		SimpleDateFormat hdsdf = new SimpleDateFormat("yyyyMM");
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		Calendar cal = Calendar.getInstance();

		String hadoopname = "";
		String filename = "";
		String localfile = "";
		String command = "";
		String cd = "";
		String user = "";
		Date yesterday = null;

		cal.set(Calendar.DATE, cal.get(Calendar.DATE) - 1);
		yesterday = cal.getTime();
		hadoopname = hdsdf.format(yesterday) + ".txt";
		filename = sdf.format(yesterday) + ".txt";

		// ---------------
		user = "root";
		localfile = "c:\\" + filename;
		cd = "/Ara/";
		file = new File(localfile);
		if (file.exists()) {
			sftp(cd, file, user);
		} else {
			file.createNewFile();
			sftp(cd, file, user);
		}
		System.out.println("1 sftp 끝");
		// ---------------
		// file.delete();
		// System.out.println("2 delete 끝");
		// ---------------
		user = "hadoop";
		command = "hadoop fs -appendToFile /Ara/" + filename + " /Ara/"+ hadoopname;
		exec(command, user);
		System.out.println("3 exec 끝");
		// ---------------
		// user = "root";
		// command = "rm -rf /Ara/"+filename;
		// exec(command,user);
		// System.out.println("4 delete 끝");
		// ---------------

		System.out.println("END");
	}

	public static void toMapReduce() throws Exception {

		List<String> list = null;
		SimpleDateFormat hdsdf = new SimpleDateFormat("yyyyMM");
		Calendar cal = Calendar.getInstance();

		String hadoopname = "";
		String command = "";
		String user = "";
		Date yesterday = null;

		cal.set(Calendar.DATE, cal.get(Calendar.DATE) - 1);
		yesterday = cal.getTime();
		hadoopname = hdsdf.format(yesterday) + ".txt";

		user = "hadoop";
		command = "hadoop jar /Ara/Ara.jar /Ara/" + hadoopname + " /reduce";
		exec(command, user);
		System.out.println("Ara.jar 끝");
		
		command = "hadoop fs -cat /reduce/part-r-00000";
		list = exec(command, user);

		FilterVisiteDao dao = new FilterVisiteDao();
		String[] data = null;
		String[] data2 = null;

		data = list.get(0).toString().split("\n");
		dao.HadoopInsert(data);
		
		System.out.println("mapreduce 끝");

	}

	private static Session getsession(String user) throws Exception {

		username = user;
		hostname = "192.168.238.128";
		password = "hadoop";

		JSch jsch = new JSch();
		Session session = jsch.getSession(username, hostname, 22);
		session.setConfig("StrictHostKeyChecking", "no");
		if (password != null) {
			session.setPassword(password);
		}
		return session;
	}

	public static List<String> exec(String command, String user) {
		List<String> ret = new ArrayList<String>();
		try {
			Session session = getsession(user);
			session.connect();
			ChannelExec channelExec = (ChannelExec) session.openChannel("exec");
			channelExec.setPty(true);
			channelExec.setCommand(command);
			InputStream inputStream = channelExec.getInputStream();
			channelExec.connect(3000);

			String output = "";
			byte[] buf = new byte[1024];
			int length;
			while ((length = inputStream.read(buf)) != -1) {
				output += new String(buf, 0, length);
			}
			ret.add(StringUtils.chop(output));
			channelExec.disconnect();
			session.disconnect();
		} catch (Exception e) {
			e.printStackTrace();
		}
		return ret;
	}

	public static void sftp(String cd, File file, String user) throws Exception {
		FileInputStream in = new FileInputStream(file);
		Session session = getsession(user);
		session.connect();

		ChannelSftp channelSftp = (ChannelSftp) session.openChannel("sftp");
		channelSftp.connect();
		channelSftp.cd(cd);
		channelSftp.put(in, file.getName());

		channelSftp.disconnect();
		session.disconnect();
	}
}